using System;
using System.Collections.Generic;
using System.Collections;
using System.Text;
using System.Runtime.Serialization.Formatters.Binary;
using System.IO;
using System.Threading;

namespace ErlyThreads
{
	public class ReceivedMessage
	{
		object mSenderId = null;
		object mData = null;

		public object SenderId
		{
			get { return mSenderId; }
		}

		public object Message
		{
			get { return mData; }
		}

		private ReceivedMessage() { }
		internal ReceivedMessage(
				object pSenderId,
				object pData)
		{
			mSenderId = pSenderId;
			mData = pData;
		}


		internal static ReceivedMessage CreateMessage(
				object pIdFrom,
				object pMessage)
		{
			using (MemoryStream stream = new MemoryStream())
			{
				// we have to force the object through serialization
				BinaryFormatter formatter = new BinaryFormatter();

				formatter.Serialize(
						stream,
						pMessage);
				stream.Position = 0;
				object message = formatter.Deserialize(stream);

				return new ReceivedMessage(
						pIdFrom,
						message);
			}
		}
	}

	public interface IProcessHandle
	{
		void DispatchQueuedMessages(
				Action<ReceivedMessage> pDistributor);
		void DispatchQueuedMessages(
				IEnumerable<Type> pTypeFilter,
				Action<ReceivedMessage> pDistributor);
		int NumberOfQueuedMessages { get; }
		object Id { get; }
		void SendMessage(
				object pRecipient,
				object pMessage);
		void SubscribeToMessages(
				object pSenderId,
				Converter<Type, bool> pDescriminator);

		/// <summary>
		/// Waits for messages indefinately
		/// </summary>
		void WaitForMessages();

		/// <summary>
		/// Waits for messages.
		/// </summary>
		/// <param name="pMillisecondsTimeout">The milliseconds timeout.</param>
		/// <returns>true if the messages arrived, false if the timeout occurred</returns>
		bool WaitForMessages(int? pMillisecondsTimeout);

		/// <summary>
		/// Waits until at least one of each of the type of the messages in the pFilter
		/// is present in the undelivered messages queue.
		/// </summary>
		/// <param name="pFilter"></param>
		void WaitForSpecificMessages(
				IEnumerable<Type> pFilter);

		/// <summary>
		/// Waits until at least one of each of the type of the messages in the pFilter
		/// is present in the undelivered messages queue.
		/// </summary>
		/// <param name="pFilter"></param>
		/// <returns>true if the messages arrived, false if the timeout occurred</returns>
		bool WaitForSpecificMessages(
			int? pMilliSecondsTimeout,
			IEnumerable<Type> pFilter);

		void WaitForOneOfMessages(
				IEnumerable<Type> pFilter);

		bool WaitForOneOfMessages(
			int? pMilleSecondsTimeout,
			IEnumerable<Type> pFilter);

		void InitProcess(
				object pId,
				Action<IProcessHandle> pStart);
		void InitProcess(
				object pId,
				Action<IProcessHandle> pStart,
				bool pHintIsShortTaks);
		void StartProcess(
				object pId);

		/// <summary>
		/// Counts the messages. returns counts by type.
		/// </summary>
		/// <param name="pMessageTypeList">The p message type list.</param>
		/// <returns></returns>
		IDictionary<Type, int> CountMessages(IEnumerable<Type> pMessageTypeList);

		/// <summary>
		/// Counts the messages. returns counts by type.
		/// </summary>
		/// <returns></returns>
		IDictionary<Type, int> CountMessages();
	}

	abstract class ThreadContainerBase : IProcessHandle
	{
		struct MessageHandle
		{
			public MessageHandle(
					byte[] pData,
					object pSender)
			{
				mMessageData = pData;
				mSender = pSender;
			}

			public byte[] mMessageData;
			public object mSender;
		}

		/// <summary>
		/// The delegate that the thread will run
		/// </summary>
		protected Action<ThreadContainerBase> mThreadStart;
		
		/// <summary>
		/// The Id of the current thread.
		/// </summary>
		private object mId;
		
		private ProcessRuntime mThreadMaster;
		
		/// <summary>
		/// The queue that will hold the messages. Use the lock to access this variable.
		/// </summary>
		/// <see cref="mMessagesLock"/>
		private Queue<ReceivedMessage> mLockedMessages =
				new Queue<ReceivedMessage>();

		/// <summary>
		/// The lock for the mLockedMessages Queue
		/// </summary>
		readonly object mMessagesLock = new object();

		private IList<AutoResetEvent> mLockedNewMessagesArrivedSignal =
				new List<AutoResetEvent>();
		
		private readonly object mNewMessagesArrivedSignalLock = new object();

		/// <summary>
		/// Gets the ThreadMaster that handles this ThreadContainer
		/// </summary>
		public ProcessRuntime ThreadMaster
		{
			get { return mThreadMaster; }
		}

		/// <summary>
		/// Gets the Id of the current thread
		/// </summary>
		public object Id
		{
			get { return mId; }
		}

		internal static ThreadContainerBase ThreadContainerBaseFactory(
				ProcessRuntime pMaster,
				Action<IProcessHandle> pStart,
				object pId,
				bool pIsShortTaskHint)
		{
			ThreadContainerBase result = null;

			if (!pIsShortTaskHint)
			{
				result = new ThreadContainer();
			}
			else
			{
				result = new ThreadPoolTypeContainer();
			}


			result.mThreadMaster = pMaster;
			result.mId = pId;
			result.mThreadStart = Delegate_Wrap(pStart);

			return result;
		}



		/// <summary>
		/// Impliment the thread's start
		/// </summary>
		internal abstract void Start();

		/// <summary>Sends a message to another thread and to the subscribers of this thread's
		/// messages</summary>
		/// <param name="pIdTo">The recipient(s) for the message.</param>
		/// <param name="pMessage">The masseg itself. Will throw if the message is not serializable.</param>
		public void Message_Send(object pIdTo, object pMessage)
		{
			mThreadMaster.Message_Send(
					mId,
					pIdTo,
					pMessage);
		}

		/// <summary>
		/// Subscribe to a filtered list of messages from a named sender
		/// </summary>
		/// <param name="pSenderId">the sender of the message's id</param>
		/// <param name="pFilter">A filter that will determine if you want
		/// to receive messages of a certain type. Pass null to indicate you
		/// are interested in all messages.</param>
		public void Messages_Subscribe(
				object pSenderId,
				Converter<Type, bool> pFilter)
		{
			mThreadMaster.Messages_Subscribe(
					pSenderId,
					Id,
					pFilter != null ?
							pFilter :
							delegate(Type t)
							{
								return true;
							});
		}

		/// <summary>
		/// Receives messages to this thread
		/// </summary>
		/// <param name="pIdFrom">The Id of the thread sending the message</param>
		/// <param name="pMessage">The message itself</param>
		internal void Message_Receive(
				ReceivedMessage pReceivedMessage)
		{
			lock (mMessagesLock)
			{
				mLockedMessages.Enqueue(pReceivedMessage);
			}

			ThreadPool.QueueUserWorkItem(delegate(object o)
			{
				lock (mNewMessagesArrivedSignalLock)
				{
					foreach (AutoResetEvent are in mLockedNewMessagesArrivedSignal)
					{
						are.Set();
					}
				}
			});
		}

		public void DispatchQueuedMessages(
				Action<ReceivedMessage> pCallback)
		{
			Queue<ReceivedMessage>
					newQueue = new Queue<ReceivedMessage>(),
					oldQueue = null;

			// replace the existing queue
			lock (mMessagesLock)
			{
				oldQueue = mLockedMessages;
				mLockedMessages = newQueue;
			}

			// dequeue all the old messages
			while (oldQueue.Count > 0)
			{
				pCallback(oldQueue.Dequeue());
			}
		}

		public void DispatchQueuedMessages(
				IEnumerable<Type> pMessageTypeFilter,
				Action<ReceivedMessage> pDistributor)
		{
			Queue<ReceivedMessage>
					oldQueue = null,
					deliveredQueue = new Queue<ReceivedMessage>();

			// replace the existing queue
			lock (mMessagesLock)
			{
				oldQueue = mLockedMessages;
				mLockedMessages = new Queue<ReceivedMessage>();

				while (oldQueue.Count > 0)
				{
					ReceivedMessage temp = oldQueue.Dequeue();

					bool inFilter = false;

					foreach (Type t in pMessageTypeFilter)
					{
						if (t.Equals(temp.Message.GetType()))
						{
							deliveredQueue.Enqueue(temp);
							inFilter = true;
							break;
						}
					}

					if (!inFilter)
					{
							mLockedMessages.Enqueue(temp);
					}
				}
			}

			while (deliveredQueue.Count > 0)
			{
				pDistributor(deliveredQueue.Dequeue());
			}
		}

		ReceivedMessage[] GetCopyOfCurrentMessagesInWaiting()
		{
			ReceivedMessage[] result = null;
			lock (mMessagesLock)
			{
				result = mLockedMessages.ToArray();
			}
			return result;
		}

		/// <summary>
		/// Waits for one of the messages indefinately
		/// </summary>
		/// <param name="pList">The list of messages that will return the call.</param>
		public void WaitForOneOfMessages(
				IEnumerable<Type> pList)
		{
			WaitForOneOfMessages(null, pList);
		}

		/// <summary>
		/// Counts the messages. returns counts by type.
		/// </summary>
		/// <returns></returns>
		public IDictionary<Type, int> CountMessages()
		{
			return CountMessages(null);
		}

		/// <summary>
		/// Counts the messages. returns counts by type.
		/// </summary>
		/// <param name="pMessageTypeList">The p message type list.</param>
		/// <returns></returns>
		public IDictionary<Type, int> CountMessages(IEnumerable<Type> pMessageTypeList)
		{
			Dictionary<Type, int> result = new Dictionary<Type, int>();

			Converter<Type, bool> inTypeList = delegate(Type t)
   		{
				if (pMessageTypeList == null) { return true;}
				foreach(Type lt in pMessageTypeList)
				{
					if (lt.Equals(t)) { return true; }
				}
   			return false;
   		};
			ReceivedMessage[] currentMessages = GetCopyOfCurrentMessagesInWaiting();
			foreach(ReceivedMessage msg in currentMessages)
			{
				Type messageType = msg.Message.GetType();
				if (inTypeList(messageType))
				{
					if (!result.ContainsKey(messageType))
					{
						result.Add(messageType, 0);
					}
					result[msg.Message.GetType()] += 1;
				}
			}

			return result;
		}

		/// <summary>
		/// Waits for one of the supplied messages for a specified number of milliseconds
		/// </summary>
		/// <param name="pMillisecondsTimeout">The milliseconds timeout.</param>
		/// <param name="pList">The list of messags that, if one of them arrives, will cause the call to return true.</param>
		/// <returns>true if one of the messages arrived in the queue, false if the wait timed-out</returns>
		public bool WaitForOneOfMessages(
			int? pMillisecondsTimeout,
			IEnumerable<Type> pList)
		{
			bool found = false;
			int searchedIndex = -1;

			DateTime earliestCallMustEnd =
				pMillisecondsTimeout.HasValue ?
				DateTime.Now + new TimeSpan(0, 0, 0, 0, pMillisecondsTimeout.Value) :
				DateTime.MaxValue;

			// lets wait until at least there are messages
			// will return true if pMilliSecondsTimeout == null
			bool gotMessages = WaitForMessages(pMillisecondsTimeout);

			// can only be false if there was a timeout, so this result is also timed-out
			if (!gotMessages) { return false; }

			// wait for the remainde of the available millisecends if set
			return WaitForMessagesCondition(
				pMillisecondsTimeout.HasValue ? (int?)(Convert.ToInt32((earliestCallMustEnd - DateTime.Now).TotalMilliseconds)) : null,
				delegate()
				{
					ReceivedMessage[] current = GetCopyOfCurrentMessagesInWaiting();
					for (int i = searchedIndex + 1; i < current.Length && !found; i++)
					{
						searchedIndex = i;

						foreach (Type t in pList)
						{
							if (t.Equals(current[i].Message.GetType()))
							{
								found = true;
								break;
							}
						}
					}

					return found;
				});
		}

		/// <summary>
		/// Waits for specific messages.
		/// </summary>
		/// <param name="pMilliSecondsTimeout">The milliseconds timeout.</param>
		/// <param name="pFilter">The filter. This call waits until all of these messages arrived, or it timed-out</param>
		/// <returns>true if the messages all arrived, false if the wait timed-out</returns>
		public bool WaitForSpecificMessages(
			int? pMillisecondsTimeout,
			IEnumerable<Type> pFilter)
		{
			DateTime earliestCallMustEnd = 
				pMillisecondsTimeout.HasValue ?
				DateTime.Now + new TimeSpan(0, 0, 0, 0, pMillisecondsTimeout.Value) : 
				DateTime.MaxValue;

			int millesecondsLeft = -1;

			foreach (Type t in pFilter)
			{
				while (true)
				{
					// check if there are any time left before the timeout
					if (pMillisecondsTimeout.HasValue)
					{
						millesecondsLeft = Convert.ToInt32((earliestCallMustEnd - DateTime.Now).TotalMilliseconds);
						if (millesecondsLeft < 0)
						{
							return false;
						}
					}

					if (WaitForOneOfMessages(
						pMillisecondsTimeout.HasValue ? (int?)millesecondsLeft : null,
						new Type[] { t }))
					{
						break;
					}
				}
			}

			// indicate that all of the messages arrived before the timeout
			return true;
		}

		/// <summary>
		/// Waits until at least one of each of the type of the messages in the pFilter
		/// is present in the undelivered messages queue.
		/// </summary>
		/// <param name="pFilter">the list of messages of which each one must arrive before the call returns</param>
		public void WaitForSpecificMessages(
				IEnumerable<Type> pFilter)
		{
			foreach (Type t in pFilter)
			{
				WaitForOneOfMessages(new Type[] { t });
			}
		}

		delegate bool WaitForMessagesConditionDelegate();

		/// <summary>
		/// Waits for messages to arrive until the condition parameter returns true
		/// </summary>
		/// <param name="pConditionMet">A condition test that causes the wait to stop.</param>
		private void WaitForMessagesCondition(
			WaitForMessagesConditionDelegate pConditionMet)
		{
			WaitForMessagesCondition(null, pConditionMet);
		}

		/// <summary>
		/// Waits for messages to arrive until the condition parameter returns true
		/// </summary>
		/// <param name="pConditionMet">A condition test that causes the wait to stop.</param>
		/// <param name="pMillisecondsTimeout">null for infinate wait, otherwise milliseconds timeout</param>
		/// <returns>true if the condition was met, false if the timeout occurred</returns>
		private bool WaitForMessagesCondition(
			int? pMillisecondsTimeout,
			WaitForMessagesConditionDelegate pConditionMet)
		{
			// earliest time that the call may end. usefull for timeout values
			DateTime earliestCallMustEnd =
				pMillisecondsTimeout.HasValue ?
				DateTime.Now + new TimeSpan(0, 0, 0, 0, pMillisecondsTimeout.Value) :
				DateTime.MaxValue;

			AutoResetEvent newEvent = null;

			// we put this in a try-catch, largely so that we can safely execute the clean-up code
			// even though we are having returns interspersed
			try
			{
				int milliSecondsRemaining = -1;

				// set to signaled, so that the test can run at least
				// once before it locks and waits for messages
				newEvent = new AutoResetEvent(false);
				lock (mNewMessagesArrivedSignalLock)
				{
					mLockedNewMessagesArrivedSignal.Add(newEvent);
				}

				// repeat until the condition parameter returns true
				// only tries the test if a message was delivered
				while (!pConditionMet())
				{
					// calculate a remaining number of milliseconds for which a call to
					// waitone may wait if a timeout was specified
					if (pMillisecondsTimeout.HasValue)
					{
						milliSecondsRemaining = Convert.ToInt32((earliestCallMustEnd - DateTime.Now).TotalMilliseconds);

						// if a timeout _was_ specified, and the time at which to quit is before the current
						// time, then return indicating that the condition was not met
						if (milliSecondsRemaining < 0)
						{
							return false;
						}
					}

					// if waitone returns false, the timeout occurred
					if (!newEvent.WaitOne(
						milliSecondsRemaining,
						false))
					{
						return false;
					}
				}

				// indicates that the condition was met
				return true;
			}
			catch (Exception)
			{
				throw;
			}
			finally
			{	
				// lets put freeing up the signal on a background thread
				// so that this one may continue with its little bit of time-slice
				// instead of blocking again
				ThreadPool.QueueUserWorkItem(delegate(object o)
				{
					lock (mNewMessagesArrivedSignalLock)
					{
						mLockedNewMessagesArrivedSignal.Remove(newEvent);
					}
				});
			}
		}


		/// <summary>
		/// allows the client to determine if any messages have been
		/// queued for receiving.
		/// </summary>
		public int NumberOfQueuedMessages
		{
			get
			{
				int result = 0;
				lock (mMessagesLock)
				{
					result = mLockedMessages.Count;
				}
				return result;
			}
		}

		/// <summary>
		/// Waits for messages indefinately
		/// </summary>
		public void WaitForMessages()
		{
			WaitForMessages(null);
		}

		/// <summary>
		/// This call blocks until new messages are received or the timeout occurs
		/// </summary>
		/// <param name="pMillicesondsTimeout">if non null, will timeout after at least this many milliseconds</param>
		/// <returns>true if the messages were received, false if the timeout occurred</returns>
		public bool WaitForMessages(int? pMillicesondsTimeout)
		{
			return WaitForMessagesCondition(
				pMillicesondsTimeout,
				delegate()
				{
					bool result = false;
					lock (mMessagesLock)
					{
						result = mLockedMessages.Count > 0;
					}
					return result;
				});
		}

		/// <summary>
		/// Wraps the thread to handle catered-for eventualities (uncaught exception etc)
		/// </summary>
		static internal Action<ThreadContainerBase> Delegate_Wrap(
				Action<IProcessHandle> pCallback)
		{
			return delegate(ThreadContainerBase pBase)
			{
				try
				{
					pBase.Message_Send(
							null,
							new Messages.ThreadStarted(
									pBase.Id));

					pCallback(pBase);
				}
				catch (Exception e)
				{
					pBase.Message_Send(
							null,
							new Messages.ThreadDied(
									pBase.Id,
									e));
				}
				finally
				{
					pBase.Message_Send(
							null,
							new Messages.ThreadStopped(
									pBase.Id));
				}
			};
		}

		#region IProcessHandle Members

		void IProcessHandle.DispatchQueuedMessages(Action<ReceivedMessage> pDistributor)
		{
			this.DispatchQueuedMessages(pDistributor);
		}

		void IProcessHandle.DispatchQueuedMessages(
				IEnumerable<Type> pTypeFilter,
				Action<ReceivedMessage> pCallback)
		{
			this.DispatchQueuedMessages(
					pTypeFilter,
					pCallback);
		}

		int IProcessHandle.NumberOfQueuedMessages
		{
			get { return this.NumberOfQueuedMessages; }
		}

		object IProcessHandle.Id
		{
			get { return this.Id; }
		}

		void IProcessHandle.SendMessage(object pRecipient, object pMessage)
		{
			this.Message_Send(
					pRecipient,
					pMessage);
		}

		void IProcessHandle.SubscribeToMessages(object pSenderId, Converter<Type, bool> pDescriminator)
		{
			this.Messages_Subscribe(
					pSenderId,
					pDescriminator);
		}

		void IProcessHandle.WaitForMessages()
		{
			this.WaitForMessages();
		}

		void IProcessHandle.InitProcess(
				object pID,
				Action<IProcessHandle> pStart)
		{
			(this as IProcessHandle).InitProcess(
					pID,
					pStart,
					false);
		}

		void IProcessHandle.InitProcess(
				object pId,
				Action<IProcessHandle> pStart,
				bool pHintIsShortTask)
		{
			this.mThreadMaster.InitProcess(
					pId,
					pStart,
					pHintIsShortTask);
		}

		void IProcessHandle.StartProcess(
				object pId)
		{
			mThreadMaster.StartProcess(pId);
		}

		void IProcessHandle.WaitForSpecificMessages(
				IEnumerable<Type> pFilter)
		{
			this.WaitForSpecificMessages(pFilter);
		}

		void IProcessHandle.WaitForOneOfMessages(
				IEnumerable<Type> pList)
		{
			this.WaitForOneOfMessages(pList);
		}

		#endregion
	}

	class ThreadContainer : ThreadContainerBase
	{
		internal override void Start()
		{
			Thread theThread = new Thread(delegate()
			{
				this.mThreadStart(this);
			});
			theThread.Start();
		}
	}

	class ThreadPoolTypeContainer : ThreadContainerBase
	{
		internal override void Start()
		{
			ThreadPool.QueueUserWorkItem(delegate(object pDummyState)
			{
				this.mThreadStart(this);
			});
		}
	}
}
